package com.flink.stream.kafka

import java.util.Properties

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.util.serialization.SimpleStringSchema

object KafkaStreamDemo {
  def main(args: Array[String]): Unit = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers","localhost:9092")
    properties.setProperty("zookeeper.connect","localhost:2181")
    properties.setProperty("group.id","test")
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val consumer=new FlinkKafkaConsumer010[String]("count-source",new SimpleStringSchema(),properties)
    val stream = env.addSource(consumer)
    stream.print()
    val conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build()
    //stream.addSink(new RedisSink[String]())
  }
}
